using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Model;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Ddl;
using Me.Prettyprint.Hector.Api.Mutation;
using Me.Prettyprint.Hector.Api.Query;
using Me.Prettyprint.Cassandra.Service.Clock;

namespace Me.Prettyprint.Hector.Api.Factory
{
    /// <summary>
    /// A convenience class with bunch of factory static methods to help 
    /// create a mutator, queries etc.
    /// </summary>
	public sealed class HFactory
	{
        private static Dictionary<string,Cluster> clusters = new Dictionary<string,Cluster>();
		private static readonly ConsistencyLevelPolicy DEFAULT_CONSISTENCY_LEVEL_POLICY = new QuorumAllConsistencyLevelPolicy();
		public HFactory()
		{
		}
        public static Cluster GetCluster(string clusterName)
        {
            lock(clusters)
            {
                return clusters[clusterName];
            }
        }
        /// <summary>
        /// Method tries to create a Cluster instance for an existing Cassandra
        /// cluster. If another class already called getOrCreateCluster, the factory
        /// returns the cached instance. If the instance doesn't exist in memory, a new
        /// ThriftCluster is created and cached.
        /// 
        /// Example usage for a default installation of Cassandra.
        /// 
        /// String clusterName = "Test Cluster"; String host = "localhost:9160";
        /// Cluster cluster = HFactory.getOrCreateCluster(clusterName, host);
        /// 
        /// Note the host should be the hostname and port number. It is preferable to
        /// use the hostname instead of the IP address.
        /// </summary>
        /// <param name="clusterName"></param>
        /// <returns></returns>
        public static Cluster GetOrCreateCluster(string clusterName)
        {
            return GetOrCreateCluster(clusterName, null,null);
        }
        /// <summary>
        /// Method tries to create a Cluster instance for an existing Cassandra
        /// cluster. If another class already called getOrCreateCluster, the factory
        /// returns the cached instance. If the instance doesn't exist in memory, a new
        /// ThriftCluster is created and cached.
        /// 
        /// Example usage for a default installation of Cassandra.
        /// 
        /// String clusterName = "Test Cluster"; String host = "localhost:9160";
        /// Cluster cluster = HFactory.getOrCreateCluster(clusterName, host);
        /// 
        /// Note the host should be the hostname and port number. It is preferable to
        /// use the hostname instead of the IP address.
        /// </summary>
        /// <param name="clusterName"></param>
        /// <param name="hostIp">host:ip format string</param>
        /// <returns></returns>
        public static Cluster GetOrCreateCluster(string clusterName, string hostIp)
        {
            return GetOrCreateCluster(clusterName, new CassandraHostConfigurator(hostIp));
        }
        /// <summary>
        /// Calls the three argument version with a null credentials map. 
        /// </summary>
        /// <param name="clusterName"></param>
        /// <param name="cassandraHostConfigurator"></param>
        /// <returns></returns>
		public static Cluster GetOrCreateCluster(System.String clusterName, CassandraHostConfigurator cassandraHostConfigurator)
		{
			return CreateCluster(clusterName, cassandraHostConfigurator, null);
		}
        /// <summary>
        /// Method tries to create a Cluster instance for an existing Cassandra cluster.
        /// If another class already called getOrCreateCluster, the factory returns the
        /// cached instance. If the instance doesn't exist in memory, a new ThriftCluster
        /// is created and cached.
        /// 
        /// Example usage for a default installation of Cassandra.
        /// string clusterName = "Test Cluster"; String host = "localhost:9160";
        /// Cluster cluster = HFactory.getOrCreateCluster(clusterName, new
        ///                     CassandraHostConfigurator(host));
        /// </summary>
        /// <param name="clusterName"></param>
        /// <param name="cassandraHostConfigurator"></param>
        /// <param name="credentials">The credentials map for authenticating a Keyspace</param>
        /// <returns></returns>
		public static Cluster GetOrCreateCluster(string clusterName, CassandraHostConfigurator cassandraHostConfigurator, Dictionary<string,string> credentials)
		{
			return CreateCluster(clusterName, cassandraHostConfigurator, credentials);
		}
        /// <summary>
        /// Method is used to create the cluster. End users should call GetOrCreateCluster
        /// method instead.
        /// </summary>
        /// <param name="clusterName"></param>
        /// <param name="cassandraHostConfigurator"></param>
        /// <param name="credentials"></param>
        /// <returns></returns>
		public static Cluster CreateCluster(string clusterName, CassandraHostConfigurator cassandraHostConfigurator, Dictionary<string,string> credentials)
        {
            Cluster c = null;
            lock(clusters)
            {

                if (!clusters.ContainsKey(clusterName))
                {
                    c = new ThriftCluster(clusterName, cassandraHostConfigurator, credentials);
                    clusters.Add(clusterName, c);
                    c.OnStartup();
                }
                else
                {
                    c = clusters[clusterName];
                }
            }
            return c;
        }
        /// <summary>
        /// Shutdown this cluster, removing it from the Map. This operation is
        /// extremely expensive and should not be done lightly.
        /// </summary>
        /// <param name="cluster"></param>
		public static void ShutdownCluster(Cluster cluster)
		{
			lock (clusters)
			{
				string clusterName = cluster.Name;
				if (clusters.ContainsKey(clusterName))
				{
					cluster.ConnectionManager.Shutdown();
					clusters.Remove(clusterName);
				}
			}
		}
        public static HColumn<N, V> CreateColumn<N, V>(N name, V value, long clock)
        {
            return new HColumnImpl<N, V>(name, 
                value, 
                clock, 
                (Serializer<N>) SerializerTypeInferer.GetSerializer<N>(name), 
                (Serializer<V>) SerializerTypeInferer.GetSerializer<V>(value));
        }
        public static HColumn<N, V> CreateColumn<N, V>(N name, V value)
        {
            return CreateColumn(name, value, CreateClock());
        }
        public static HColumn<N, V> CreateColumn<N, V>(N name, V value, long clock, int ttl,
            Serializer<N> nameSerializer, Serializer<V> valueSerializer)
        {
            return new HColumnImpl<N, V>(name, value, clock, ttl, nameSerializer,
                valueSerializer);
        }
        public static HColumn<N, V> CreateColumn<N, V>(N name, V value,
            Serializer<N> nameSerializer, Serializer<V> valueSerializer)
        {
            return new HColumnImpl<N, V>(name, value, CreateClock(), nameSerializer,
                valueSerializer);
        }
        public static HColumn<N, V> CreateColumn<N, V>(N name, V value, int ttl,
            Serializer<N> nameSerializer, Serializer<V> valueSerializer)
        {
            return new HColumnImpl<N, V>(name, value, CreateClock(), ttl, nameSerializer,
                valueSerializer);
        }
        public static ColumnFamilyDefinition CreateColumnFamilyDefinition(string keyspace,
            string cfName, ComparatorType comparatorType)
        {
            return new ThriftCfDef(keyspace, cfName, comparatorType);
        }
        public static ColumnFamilyDefinition CreateColumnFamilyDefinition(string keyspace,
            string cfName, ComparatorType comparatorType,
            List<ColumnDefinition> columnMetadata)
        {
            return new ThriftCfDef(keyspace, cfName, comparatorType, columnMetadata);
        }
        public static ColumnFamilyDefinition CreateColumnFamilyDefinition(string keyspace, string cfName)
        {
            return new ThriftCfDef(keyspace, cfName);
        }
        /// <summary>
        /// Creates a Keyspace with the default consistency level policy.
        /// 
        /// Example usage.
        /// 
        /// string clusterName = "Test Cluster"; 
        /// string host = "localhost:9160";
        /// Cluster cluster = HFactory.getOrCreateCluster(clusterName, host); 
        /// string keyspaceName = "testKeyspace";
        /// Keyspace myKeyspace = HFactory.createKeyspace(keyspaceName, cluster);
        /// </summary>
        /// <param name="keyspace"></param>
        /// <param name="cluster"></param>
        /// <returns></returns>
		public static Keyspace CreateKeyspace(string keyspace, Cluster cluster)
		{
			return CreateKeyspace(keyspace, cluster, CreateDefaultConsistencyLevelPolicy(), FailoverPolicy.ON_FAIL_TRY_ALL_AVAILABLE);
		}
		/// <summary>
		/// Creates a Keyspace with the given consistency level. For a reference
        /// to the consistency level, please refer to http://wiki.apache.org/cassandra/API.
		/// </summary>
		/// <param name="keyspace"></param>
		/// <param name="cluster"></param>
		/// <param name="consistencyLevelPolicy"></param>
		/// <returns></returns>
		public static Keyspace CreateKeyspace(string keyspace, Cluster cluster, ConsistencyLevelPolicy consistencyLevelPolicy)
		{
			return CreateKeyspace(keyspace, cluster, consistencyLevelPolicy, FailoverPolicy.ON_FAIL_TRY_ALL_AVAILABLE);
		}
		/// <summary>
		/// Creates a Keyspace with the given consistency level. For a reference
        /// to the consistency level, please refer to http://wiki.apache.org/cassandra/API.
		/// </summary>
		/// <param name="keyspace"></param>
		/// <param name="cluster"></param>
		/// <param name="consistencyLevelPolicy"></param>
		/// <param name="failoverPolicy"></param>
		/// <returns></returns>
		public static Keyspace CreateKeyspace(string keyspace, Cluster cluster, ConsistencyLevelPolicy consistencyLevelPolicy, FailoverPolicy failoverPolicy)
		{
			return new ExecutingKeyspace(keyspace, cluster.ConnectionManager, consistencyLevelPolicy, failoverPolicy, cluster.Credentials);
		}
        /// <summary>
        /// Use createKeyspaceDefinition to add a new Keyspace to cluster. Example:
        /// 
        /// string testKeyspace = "testKeyspace";
        /// KeyspaceDefinition newKeyspace = HFactory.createKeyspaceDefinition(testKeyspace);
        /// cluster.addKeyspace(newKeyspace);
        /// </summary>
        /// <param name="keyspace"></param>
        /// <returns></returns>
        public static KeyspaceDefinition CreateKeyspaceDefinition(string keyspace)
        {
            return new ThriftKsDef(keyspace);
        }
        /// <summary>
        /// Create a new KeyspaceDefinition with a list of ColumnFamilies.
        /// 
        /// </summary>
        /// <param name="keyspaceName"></param>
        /// <param name="strategyClass"></param>
        /// <param name="replicationFactor"></param>
        /// <param name="cfDefs"></param>
        /// <returns></returns>
        public static KeyspaceDefinition CreateKeyspaceDefinition(string keyspaceName, string strategyClass, 
            int replicationFactor, List<ColumnFamilyDefinition> cfDefs)
        {
            return new ThriftKsDef(keyspaceName, strategyClass, replicationFactor, cfDefs);
        }
        /// <summary>
        /// Creates a mutator for updating records in a keyspace.
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <returns></returns>
		public static Mutator<K> CreateMutator<K>(Keyspace keyspace, Serializer<K> keySerializer)
        {
            return new MutatorImpl<K>(keyspace, keySerializer);
        }
        public static Mutator<K> CreateMutator<K, N, V>(Keyspace keyspace, Serializer<K> keySerializer, BatchSizeHint sizeHint)
        {
            return new MutatorImpl<K>(keyspace, keySerializer, sizeHint);
        }
        /// <summary>
        /// Creates a new ThriftColumnQuery with the given keyspace, and serializers.
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <typeparam name="V"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <param name="valueSerializer"></param>
        /// <returns></returns>
        public static ColumnQuery<K, N, V> CreateColumnQuery<K, N, V>(Keyspace keyspace, Serializer<K> keySerializer,
            Serializer<N> nameSerializer, Serializer<V> valueSerializer)
        {
            return new ThriftColumnQuery<K, N, V>(keyspace, keySerializer, nameSerializer, valueSerializer);
        }
        /// <summary>
        /// Creates a new ThriftCounterColumnQuery with the given keyspace and
        /// serializers.
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <returns></returns>
        public static CounterQuery<K, N> CreateCounterColumnQuery<K, N>(Keyspace keyspace, Serializer<K> keySerializer, 
            Serializer<N> nameSerializer)
        {
            return new ThriftCounterColumnQuery<K, N>(keyspace, keySerializer, nameSerializer);
        }
        /// <summary>
        /// Creates a new ThriftSliceCounterQuery for the given keyspace, and name type
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <returns></returns>
        public static SliceCounterQuery<K, N> CreateCounterSliceQuery<K, N>(Keyspace keyspace, 
            Serializer<K> keySerializer, Serializer<N> nameSerializer)
        {
            return new ThriftSliceCounterQuery<K, N>(keyspace, keySerializer, nameSerializer);
        }
        /// <summary>
        /// Creates a ThriftCountQuery for the given keyspace and name type
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <returns></returns>
        public static CountQuery<K, N> CreateCountQuery<K, N>(Keyspace keyspace,
            Serializer<K> keySerializer, Serializer<N> nameSerializer)
        {
            return new ThriftCountQuery<K, N>(keyspace, keySerializer, nameSerializer);
        }
        /// <summary>
        /// Creates a ThriftMultigetSliceQuery for the given keyspace, name type and
        /// value type.
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <typeparam name="V"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <param name="valueSerializer"></param>
        /// <returns></returns>
        public static MultigetSliceQuery<K, N, V> CreateMultigetSliceQuery<K, N, V>(
            Keyspace keyspace, Serializer<K> keySerializer, Serializer<N> nameSerializer, Serializer<V> valueSerializer)
        {
            return new ThriftMultigetSliceQuery<K, N, V>(keyspace, keySerializer, nameSerializer, valueSerializer);
        }
        /// <summary>
        /// Creates a ThriftMultigetSliceCounterQuery for the given keyspace with the
        /// given name type
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <returns></returns>
        public static MultigetSliceCounterQuery<K, N> CreateMultigetSliceCounterQuery<K, N>(
            Keyspace keyspace, Serializer<K> keySerializer, Serializer<N> nameSerializer)
        {
            return new ThriftMultigetSliceCounterQuery<K, N>(keyspace, keySerializer, nameSerializer);
        }
        /// <summary>
        /// Creates a ThriftSuperCountQuery for the given keyspace with the super name type
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="SN"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="superNameSerializer"></param>
        /// <returns></returns>
        public static SuperCountQuery<K, SN> CreateSuperCountQuery<K, SN>(Keyspace keyspace,
            Serializer<K> keySerializer, Serializer<SN> superNameSerializer)
        {
            return new ThriftSuperCountQuery<K, SN>(keyspace, keySerializer, superNameSerializer);
        }
        /// <summary>
        /// Creates a ThriftSubCountQuery for the given keyspace with the super name and name type
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="SN"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="superNameSerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <returns></returns>
        public static SubCountQuery<K, SN, N> CreateSubCountQuery<K, SN, N>(Keyspace keyspace, 
            Serializer<K> keySerializer, Serializer<SN> superNameSerializer, 
            Serializer<N> nameSerializer)
        {
            return new ThriftSubCountQuery<K, SN, N>(keyspace, keySerializer, superNameSerializer, nameSerializer);
        }
        /// <summary>
        /// Creates a ThriftSubSliceQuery for the given keyspace super name, name and value type
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="SN"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <typeparam name="V"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="sNameSerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <param name="valueSerializer"></param>
        /// <returns></returns>
        public static SubSliceQuery<K, SN, N, V> CreateSubSliceQuery<K, SN, N, V>(Keyspace keyspace, 
            Serializer<K> keySerializer, Serializer<SN> sNameSerializer, 
            Serializer<N> nameSerializer, Serializer<V> valueSerializer) 
        {
            return new ThriftSubSliceQuery<K, SN, N, V>(keyspace, keySerializer,
                sNameSerializer, nameSerializer, valueSerializer);
        }
        /// <summary>
        /// Creates a ThriftRangeSlicesQuery for the given keyspace name and value type
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <typeparam name="V"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <param name="valueSerializer"></param>
        /// <returns></returns>
        public static RangeSlicesQuery<K, N, V> CreateRangeSlicesQuery<K, N, V>(Keyspace keyspace, 
            Serializer<K> keySerializer, Serializer<N> nameSerializer, 
            Serializer<V> valueSerializer)
        {
            return new ThriftRangeSlicesQuery<K, N, V>(keyspace, keySerializer,
                nameSerializer, valueSerializer);
        }
        /// <summary>
        /// Creates a ThriftRangeSlicesCounterQuery for the keyspace and name type
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <returns></returns>
        public static RangeSlicesCounterQuery<K, N> CreateRangeSlicesCounterQuery<K, N>(Keyspace keyspace, 
            Serializer<K> keySerializer, Serializer<N> nameSerializer)
        {
            return new ThriftRangeSlicesCounterQuery<K, N>(keyspace, keySerializer,
                nameSerializer);
        }
        /// <summary>
        /// Creates a ThriftRangeSubSlicesQuery for the given keyspace, super name, name and value type
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="SN"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <typeparam name="V"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="sNameSerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <param name="valueSerializer"></param>
        /// <returns></returns>
        public static RangeSubSlicesQuery<K, SN, N, V> CreateRangeSubSlicesQuery<K, SN, N, V>(Keyspace keyspace, 
            Serializer<K> keySerializer, Serializer<SN> sNameSerializer, Serializer<N> nameSerializer,
            Serializer<V> valueSerializer)
        {
            return new ThriftRangeSubSlicesQuery<K, SN, N, V>(keyspace, keySerializer, sNameSerializer, nameSerializer, valueSerializer);
        }
        /// <summary>
        /// Creates ThriftRangeSubSlicesCounterQuery for the given keyspace super name and name type
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="SN"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="sNameSerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <returns></returns>
        public static RangeSubSlicesCounterQuery<K, SN, N> CreateRangeSubSlicesCounterQuery<K, SN, N>(Keyspace keyspace, 
            Serializer<K> keySerializer, Serializer<SN> sNameSerializer, 
            Serializer<N> nameSerializer)
        {
            return new ThriftRangeSubSlicesCounterQuery<K, SN, N>(keyspace, keySerializer, sNameSerializer, nameSerializer);
        }
        /// <summary>
        /// Creates a ThirftSliceQuery for the given keyspace name and value type
        /// </summary>
        /// <typeparam name="K"></typeparam>
        /// <typeparam name="N"></typeparam>
        /// <typeparam name="V"></typeparam>
        /// <param name="keyspace"></param>
        /// <param name="keySerializer"></param>
        /// <param name="nameSerializer"></param>
        /// <param name="valueSerializer"></param>
        /// <returns></returns>
        public static SliceQuery<K, N, V> CreateSliceQuery<K, N, V>(Keyspace keyspace, 
            Serializer<K> keySerializer, Serializer<N> nameSerializer, Serializer<V> valueSerializer)
        {
            return new ThriftSliceQuery<K, N, V>(keyspace, keySerializer, nameSerializer, valueSerializer);
        }
		public static ConsistencyLevelPolicy CreateDefaultConsistencyLevelPolicy()
		{
			return DEFAULT_CONSISTENCY_LEVEL_POLICY;
		}
        public static long CreateClock()
        {
            return CassandraHostConfigurator.GetClockResolution().CreateClock();
        }

        /**
         * Create a clock resolution based on <code>clockResolutionName</code> which
         * has to match any of the constants defined at {@link ClockResolution}
         * 
         * @param clockResolutionName
         *          type of clock resolution to create
         * @return a ClockResolution
         */
        public static ClockResolution CreateClockResolution(String clockResolutionName)
        {
            if (clockResolutionName.Equals(ClockResolutionFields.SECONDS))
            {
                return new SecondsClockResolution();
            }
            else if (clockResolutionName.Equals(ClockResolutionFields.MILLISECONDS))
            {
                return new MillisecondsClockResolution();
            }
            else if (clockResolutionName.Equals(ClockResolutionFields.MICROSECONDS))
            {
                return new MicrosecondsClockResolution();
            }
            else if (clockResolutionName.Equals(ClockResolutionFields.MICROSECONDS_SYNC))
            {
                return new MicrosecondsSyncClockResolution();
            }
            throw new System.ArgumentException(String.Format(
                "Unsupported clock resolution: %s", clockResolutionName));
        }
	}
}